package com.bw.gmall.realtime.app.dwd;


import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.app.func.DimAsyncFunction;
import com.bw.gmall.realtime.utils.DateFormatUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

// select   *   from  五表  where  type='insert'
public class DwdTradeOrderDetail_shop {

    public static void main(String[] args) throws Exception {

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); //生产环境中设置为Kafka主题的分区数
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        
        DataStreamSource<String> dataStreamSource = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_trade_order_detail", "DwdTradeOrderDetail_shop"));
        SingleOutputStreamOperator<JSONObject> map = dataStreamSource.map(t->JSONObject.parseObject(t));
        SingleOutputStreamOperator<JSONObject> ts = map.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                String string = element.getString("create_time");
                // 转换为毫秒级时间戳
                long time = DateFormatUtil.toTs(string,true);
                return time;
            }
        }));
        // 使用更高的并行度和更长的超时时间
        SingleOutputStreamOperator<JSONObject> jsonObjectSingleOutputStreamOperator = AsyncDataStream.unorderedWait(ts, new DimAsyncFunction<JSONObject>("DIM_SKU_INFO_ZYJ1") {
            @Override
            public String getKey(JSONObject input) {
                return input.getString("sku_id");
            }

            @Override
            public void join(JSONObject input, JSONObject dimInfo) throws Exception {
                input.put("shop", dimInfo.getString("SHOP_ID"));
            }
        }, 100, TimeUnit.SECONDS);
//         jsonObjectSingleOutputStreamOperator.print();
        // 将流存入kafka
//        jsonObjectSingleOutputStreamOperator.map(JSONObject::toString).addSink(MyKafkaUtil.getFlinkKafkaProducer("dwd_trade_order_detail_with_shop"));
        //TODO 6.启动任务
        env.execute("DwdTradeOrderDetailShop");


    }

}